package com.yc.system.service.impl;

import com.yc.common.core.domain.model.EmailCaptchaMessage;
import com.yc.common.core.redis.RedisCache;
import com.yc.system.service.EmailService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.MessagingException;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.imageio.ImageIO;
import java.awt.image.BufferedImage;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.Base64;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * @Author YanChen
 * @Date 2025/4/24 10:53
 **/
@Service
public class EmailServiceImpl implements EmailService, RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback  {

    private static final Logger logger = LoggerFactory.getLogger(EmailServiceImpl.class);

    // 最大重试次数
    private static final int MAX_RETRY_COUNT = 3;
    // 重试间隔（秒）
    private static final int RETRY_INTERVAL = 5;
    // Redis中存储重试次数的Key前缀
    private static final String RETRY_COUNT_KEY_PREFIX = "email:retry:count:";
    // Redis中存储消息内容的Key前缀
    private static final String RETRY_MESSAGE_KEY_PREFIX = "email:retry:message:";
    // Redis中Key的过期时间（分钟）
    private static final Long RETRY_KEY_EXPIRATION = 30L;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedisCache redisCache;

    @Value("${spring.mail.username}")
    private String from;

    /**
     * 初始化RabbitTemplate
     */
    @PostConstruct
    public void initRabbitTemplate() {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

    @Override
    public void sendCaptchaEmailAsync(String to, String subject, BufferedImage captchaImage) {
        try {
            // 将图片转为Base64字符串
            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
            ImageIO.write(captchaImage, "JPEG", outputStream);
            byte[] imageBytes = outputStream.toByteArray();
            String base64Image = Base64.getEncoder().encodeToString(imageBytes);

            // 创建消息
            EmailCaptchaMessage message = new EmailCaptchaMessage(to, base64Image);
            message.setSubject(subject);

            // 创建关联数据用于消息确认
            String correlationId = UUID.randomUUID().toString();
            CorrelationData correlationData = new CorrelationData(correlationId);

            // 将消息内容存入Redis以便重发
            saveMessageToRedis(correlationId, message, "email.captcha.exchange", "email.captcha.routingKey");

            // 添加消息属性
            MessagePostProcessor messagePostProcessor = msg -> {
                MessageProperties properties = msg.getMessageProperties();
                properties.setMessageId(correlationId);
                properties.setContentEncoding("UTF-8");
                properties.setPriority(5); // 设置优先级
                properties.setExpiration("60000"); // 60秒过期
                return msg;
            };

            // 发送消息到队列，使用确认机制
            rabbitTemplate.convertAndSend(
                "email.captcha.exchange",
                "email.captcha.routingKey",
                message,
                messagePostProcessor,
                correlationData
            );

            logger.info("验证码邮件消息已发送到队列，收件人：{}，消息ID：{}", to, correlationId);
        } catch (IOException e) {
            logger.error("发送验证码邮件消息到队列失败", e);
            throw new RuntimeException("发送验证码邮件消息到队列失败", e);
        }
    }

    /**
     * 将消息保存到Redis，用于消息重发
     */
    private void saveMessageToRedis(String messageId, Object message, String exchange, String routingKey) {
        // 保存消息内容
        String messageKey = RETRY_MESSAGE_KEY_PREFIX + messageId;
        redisCache.setCacheObject(messageKey, new MessageWrapper(message, exchange, routingKey), RETRY_KEY_EXPIRATION, TimeUnit.MINUTES);

        // 初始化重试次数为0
        String countKey = RETRY_COUNT_KEY_PREFIX + messageId;
        redisCache.setCacheObject(countKey, 0, RETRY_KEY_EXPIRATION, TimeUnit.MINUTES);

        logger.debug("消息已保存到Redis，准备潜在重发，消息ID: {}", messageId);
    }

    /**
     * 消息发送到交换机的回调
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (correlationData != null) {
            String messageId = correlationData.getId();
            if (ack) {
                logger.info("消息成功发送到交换机，ID: {}", messageId);
                // 删除Redis中的消息，不再需要重发
                cleanupMessageFromRedis(messageId);
            } else {
                logger.error("消息发送到交换机失败，ID: {}, 原因: {}", messageId, cause);
                // 消息发送失败，尝试重发
                retryMessage(correlationData);
            }
        }
    }

    /**
     * 消息从交换机路由到队列失败的回调
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        String messageId = message.getMessageProperties().getMessageId();
        logger.error("消息从交换机路由到队列失败，ID: {}, 交换机: {}, 路由键: {}, 回复码: {}, 回复文本: {}",
                    messageId, exchange, routingKey, replyCode, replyText);

        // 创建关联数据用于重发
        CorrelationData correlationData = new CorrelationData(messageId);
        retryMessage(correlationData);
    }

    /**
     * 消息重发逻辑
     * 使用Redis记录重发次数，超过阈值后通知人工处理
     */
    private void retryMessage(CorrelationData correlationData) {
        String messageId = correlationData.getId();
        String countKey = RETRY_COUNT_KEY_PREFIX + messageId;
        String messageKey = RETRY_MESSAGE_KEY_PREFIX + messageId;

        // 获取当前重试次数
        Integer retryCount = redisCache.getCacheObject(countKey);
        if (retryCount == null) {
            logger.warn("无法获取消息重试次数，可能已过期，消息ID: {}", messageId);
            return;
        }

        // 获取消息内容
        MessageWrapper messageWrapper = redisCache.getCacheObject(messageKey);
        if (messageWrapper == null) {
            logger.warn("无法获取消息内容，可能已过期，消息ID: {}", messageId);
            return;
        }

        // 判断是否超过最大重试次数
        if (retryCount >= MAX_RETRY_COUNT) {
            logger.error("消息重发次数已达到上限({})，放弃重发，消息ID: {}", MAX_RETRY_COUNT, messageId);
            // 可以在这里发送告警通知
            sendAlertForFailedMessage(messageId, messageWrapper);
            // 清理Redis中的消息数据
            cleanupMessageFromRedis(messageId);
            return;
        }

        // 更新重试次数
        redisCache.setCacheObject(countKey, retryCount + 1, RETRY_KEY_EXPIRATION, TimeUnit.MINUTES);

        try {
            logger.info("准备重发消息，ID: {}, 当前重试次数: {}/{}", messageId, retryCount + 1, MAX_RETRY_COUNT);

            // 创建新的关联数据
            CorrelationData newCorrelationData = new CorrelationData(messageId);

            // 添加消息属性
            MessagePostProcessor messagePostProcessor = msg -> {
                MessageProperties properties = msg.getMessageProperties();
                properties.setMessageId(messageId);
                properties.setContentEncoding("UTF-8");
                properties.setPriority(5);
                properties.setExpiration("60000");
                return msg;
            };

            // 重发消息
            rabbitTemplate.convertAndSend(
                messageWrapper.getExchange(),
                messageWrapper.getRoutingKey(),
                messageWrapper.getMessage(),
                messagePostProcessor,
                newCorrelationData
            );

            logger.info("消息重发成功，ID: {}, 重试次数: {}/{}", messageId, retryCount + 1, MAX_RETRY_COUNT);
        } catch (Exception e) {
            logger.error("消息重发失败，ID: {}, 错误: {}", messageId, e.getMessage(), e);
            // 如果是最后一次重试失败，发送告警通知
            if (retryCount + 1 >= MAX_RETRY_COUNT) {
                sendAlertForFailedMessage(messageId, messageWrapper);
                cleanupMessageFromRedis(messageId);
            }
        }
    }

    /**
     * 发送失败消息告警通知
     */
    private void sendAlertForFailedMessage(String messageId, MessageWrapper messageWrapper) {
        // 这里可以实现告警逻辑，例如发送短信、钉钉、邮件等
        // 例如可以记录到专门的数据库表中，或者调用告警API
        logger.error("发送告警通知：消息发送彻底失败，需要人工处理。消息ID: {}, 交换机: {}, 路由键: {}, 消息内容: {}",
                messageId, messageWrapper.getExchange(), messageWrapper.getRoutingKey(), messageWrapper.getMessage());
    }

    /**
     * 清理Redis中的消息数据
     */
    private void cleanupMessageFromRedis(String messageId) {
        String countKey = RETRY_COUNT_KEY_PREFIX + messageId;
        String messageKey = RETRY_MESSAGE_KEY_PREFIX + messageId;

        redisCache.deleteObject(countKey);
        redisCache.deleteObject(messageKey);

        logger.debug("已清理Redis中的消息数据，消息ID: {}", messageId);
    }

    /**
     * 消息包装类，用于在Redis中存储消息及其路由信息
     */
    private static class MessageWrapper {
        private final Object message;
        private final String exchange;
        private final String routingKey;

        public MessageWrapper(Object message, String exchange, String routingKey) {
            this.message = message;
            this.exchange = exchange;
            this.routingKey = routingKey;
        }

        public Object getMessage() {
            return message;
        }

        public String getExchange() {
            return exchange;
        }

        public String getRoutingKey() {
            return routingKey;
        }
    }
}